@synalinks_export(
[
"synalinks.modules.RecursiveLanguageModelAgent",
"synalinks.RecursiveLanguageModelAgent",
"synalinks.modules.RLM",
"synalinks.RLM",
]
)
class RecursiveLanguageModelAgent(FunctionCallingAgent):
"""A recursive-language-model agent.
Each turn the LM calls a single tool,
``run_python_code(code: str)``, which runs the snippet in a
persistent REPL sandbox (by default a ``MirageSandbox``) and
returns ``{"stdout", "stderr", "error"}``. State (variables, imports,
function definitions) accumulates across turns so the agent can build up
intermediate values, probe data, and iterate. ``submit``, the recursive
helpers and any user tools are **not** exposed to the LM as tools: they
live *inside* the sandbox as plain synchronous functions (advertised
through the tools catalog), reachable only from the code passed to
``run_python_code``.
When ``recursive=True`` (the default), two extra helpers are exposed
inside the sandbox: ``llm_query(prompt)`` and
``llm_query_batched(prompts)``. The agent then treats long inputs as
an *external environment*, it writes Python that slices, filters, and
aggregates the data, and recursively delegates semantic work to a
sub-LM only on the snippets it cares about. Compared to feeding a
long document straight into the primary LM, this trades a single
huge context for many small ones, which both fits inside provider
limits and reduces the chance of long-context regressions.
When ``recursive=False``, the agent runs without the sub-LM helpers,
useful when the task is purely computational and recursion would
only add cost.
Bound user tools (if any) appear inside the sandbox as global functions;
scripts call them directly, ``result = tool_name(...)``.
Termination: the snippet calls the in-sandbox ``submit(result=...)``
callable with the final payload. If ``max_iterations`` is reached
without ``submit``, a final inference step formats the accumulated
trajectory into the target ``schema`` / ``data_model``. Empty snippets
are not termination signals, the loop feeds back a reminder and keeps
going.
The ``llm_query`` quota is per-call: every invocation of this agent
gets a fresh budget of ``max_llm_calls`` sub-LM queries, and
concurrent invocations of the *same* agent instance each get an
independent budget: the counter and lock are built inside
``call()`` and never shared across runs.
Example:
```python
import synalinks
import asyncio
class Doc(synalinks.DataModel):
text: str
class Answer(synalinks.DataModel):
answer: str
async def main():
primary = synalinks.LanguageModel(model="openai/gpt-4o")
cheap = synalinks.LanguageModel(model="openai/gpt-4o-mini")
inputs = synalinks.Input(data_model=Doc)
outputs = await synalinks.RLM(
data_model=Answer,
language_model=primary,
sub_language_model=cheap,
max_iterations=8,
max_llm_calls=20,
)(inputs)
agent = synalinks.Program(inputs=inputs, outputs=outputs)
long_text = open("book.txt").read()
result = await agent(Doc(text=long_text))
print(result.prettify_json())
if __name__ == "__main__":
asyncio.run(main())
```
References:
- [Recursive Language Models](https://arxiv.org/abs/2512.24601)
Args:
schema (dict): Optional. The target JSON schema for the final
structured answer. If not provided, use ``data_model`` to
infer it. When both are omitted, the agent runs in
**schemaless** mode, the final generator emits a
``ChatMessage`` that is appended to the trajectory, and
``call`` returns the ``ChatMessages`` trajectory directly.
data_model (DataModel | SymbolicDataModel | JsonDataModel):
Optional. The target data model for the final answer.
language_model (LanguageModel): The language model driving the
per-turn code generator and the final-formatting step.
sub_language_model (LanguageModel): Optional. The language
model used by ``llm_query`` and ``llm_query_batched`` when
``recursive=True``, and by spawned subagents. Defaults to
``language_model``, pass a cheaper / smaller model here when
the recursive sub-queries don't need the primary LM's full
capability. Ignored when ``recursive=False``.
prompt_template (str): Optional. Prompt template forwarded to
the per-turn code generator.
examples (list): Optional. Examples forwarded to the per-turn
code generator.
instructions (str): Optional. Instructions for the per-turn
code generator. Defaults to either
`get_recursive_instructions` (when ``recursive=True``,
with the ``{max_llm_calls}`` placeholder substituted) or
`get_default_instructions` otherwise.
final_instructions (str): Optional. Instructions for the final
answer generator. Defaults to ``instructions``.
temperature (float): Optional. Sampling temperature
(Default 0.0).
max_tokens (int): Optional. Maximum number of tokens to generate.
Default None (the model's own default; caps generation length).
top_p (float): Optional. Nucleus sampling probability. Default None
(the model's own default).
top_k (int): Optional. Top-k sampling cutoff. Default None (the
model's own default).
use_inputs_schema (bool): Optional. Feed the input schema to
the generator prompt (Default False).
use_outputs_schema (bool): Optional. Feed the output schema to
the generator prompt (Default False).
reasoning_effort (str): Optional. One of ``'minimal'``,
``'low'``, ``'medium'``, ``'high'``, ``'disable'``,
``'none'``, ``None``. Default ``None``.
use_chain_of_thought (bool): Optional. Wrap the per-turn
generator in ChainOfThought so it emits a ``thinking``
field alongside the tool call. Default ``False``.
tools (list): Optional. Extra `Tool` instances exposed to
the sandbox in addition to ``submit`` (and ``llm_query`` /
``llm_query_batched`` when ``recursive=True``). The names
``submit``, ``llm_query``, and ``llm_query_batched`` are
always reserved at construction time, even when
``recursive=False``, so tool naming stays stable across the
two modes.
**Naming gotcha**: each tool is registered under
``tool.name == tool._func.__name__``. ``Tool(_my_helper)``
shows up inside the script as ``_my_helper``. Rename the
function rather than relying on an alias.
sandbox_tools (list): Optional. Explicit spelling of ``tools``;
the two are concatenated. Pass the pair
``sandbox_tools=`` / ``native_tools=`` when an agent has some
of each and ``tools=`` would read ambiguously.
native_tools (list): Optional. `Tool` instances the LM calls
**directly**, alongside ``run_python_code``, rather than from
inside a snippet. Use this for a tool the LM *consults* —
one whose answer shapes the snippet it is about to write,
like a library or documentation lookup. Keep a tool in the
sandbox when a snippet *composes* with it: when its result
feeds the next line of Python. Native tools are dispatched
with the REPL idle, are absent from the sandbox namespace and
from the tools catalog, and may not share a name with a
sandbox tool.
autonomous (bool): Optional. If ``True`` (default), run the
full code/execute/observe loop until the LM calls
``submit`` or ``max_iterations`` is reached, then produce a
structured final answer. If ``False``, require a
``ChatMessages`` input and execute a single code turn per
call, returning the updated trajectory, suitable for
human-in-the-loop use. For cross-call REPL state in
interactive mode, hand a ``Sandbox`` to ``call`` via the
``sandbox`` kwarg; the agent itself stays stateless.
return_inputs_with_trajectory (bool): Optional. Whether to
return the full trajectory alongside the final answer
(Default ``True``).
max_iterations (int): Maximum number of code-execution turns
before forcing the final answer step (Default 20).
timeout (int): Per-turn execution budget in seconds
(Default 60). Recursive sub-LM calls dominate per-turn wall
time; ``llm_query_batched`` of even a handful of prompts
can take several seconds. Snippets that exceed the budget
turn into an observation so the LM can recover on the next
turn.
recursive (bool): Optional. If ``True`` (default), expose
``llm_query`` and ``llm_query_batched`` inside the sandbox
and use the recursive instructions. If ``False``, run
without the sub-LM helpers.
max_llm_calls (int): Hard cap on sub-LM calls per agent
invocation, shared between ``llm_query`` and
``llm_query_batched`` (Default 50). Once the budget is
spent, further calls return an error string instead of a
response so the LM can fall back to code-side aggregation.
Ignored when ``recursive=False``.
max_output_chars (int): Maximum characters to include from
REPL output in the per-turn observation (Default 10_000).
Anything beyond is truncated with a
``… (truncated, N chars omitted)`` marker so a single
noisy turn cannot blow up the trajectory.
workdir (str): Optional. Host directory the agent operates on. When
building its own sandbox (i.e. no ``sandbox`` instance is supplied),
the workdir seeds the sandbox filesystem. If it contains an
``AGENTS.md`` file, its contents are also injected as an additional
input so the agent follows the declared project conventions (see
``read_agents_md``). Must point to an existing directory. Defaults
to ``None``.
skills (list): Optional. Folder paths (Agent Skill roots) whose skills
are listed for the agent as an ``<available_skills>`` context message
(see `FunctionCallingAgent`). The skill files must also be reachable
from the agent's sandbox (e.g. under ``workdir``) for their bodies to
be read on demand. Defaults to ``None``.
sandbox (Sandbox): Optional. A pre-built ``Sandbox`` instance to
reuse across calls. When supplied, the agent will not build
its own sandbox at ``call()`` time and ``sandbox_type`` is
derived from ``type(sandbox)``. Pass this when the caller
owns the sandbox lifecycle (e.g. interactive sessions where
REPL state must persist across calls). When omitted, a
fresh sandbox of ``sandbox_type`` is built per call.
sandbox_type (type): Optional. The ``Sandbox`` subclass to
instantiate when no sandbox is supplied (here or to
``call()``). Defaults to ``MirageSandbox``, or to
``type(sandbox)`` when ``sandbox`` is given. Any ``Sandbox``
subclass whose ``__init__`` accepts
``(timeout=..., name=...)`` works; register custom
subclasses with ``@register_synalinks_serializable`` so
they round-trip through ``get_config`` / ``from_config``.
max_subagent_depth (int): When ``> 0``, the agent gains
``spawn_subagents`` / ``merge_subagent`` / ``discard_subagent``
tools (called between snippets, with the REPL idle, so they can
fork it). Each subagent runs in parallel on a
`Sandbox.fork` that inherits this agent's current REPL
state (variables, functions, imports) *and* files; its work only
lands on an explicit ``merge_subagent``. ``1`` (recommended) lets
this agent spawn subagents that cannot themselves spawn; higher
values allow nesting. Defaults to ``0`` (disabled).
Across parallel subagents you can fold back **all** their file
changes, but only **one** subagent's REPL namespace per
``spawn_subagents`` batch (via ``merge_subagent(..., adopt_repl=
True)``): the REPL serializes only as a whole, so parallel
namespaces can't be unioned. That is a backend constraint, not a
design shortcut.
name (str): Optional. The name of the module.
description (str): Optional. The description of the module.
"""
def __init__(
self,
*,
schema=None,
data_model=None,
language_model=None,
sub_language_model=None,
prompt_template=None,
examples=None,
instructions=None,
final_instructions=None,
temperature=None,
max_tokens=None,
top_p=None,
top_k=None,
use_inputs_schema=False,
use_outputs_schema=False,
reasoning_effort=None,
use_chain_of_thought=False,
tools=None,
sandbox_tools=None,
native_tools=None,
autonomous=True,
return_inputs_with_trajectory=True,
max_iterations=20,
streaming=False,
timeout=60,
recursive=True,
max_llm_calls=50,
max_output_chars=10_000,
workdir=None,
skills=None,
sandbox=None,
sandbox_type=None,
max_subagent_depth=0,
_subagent_depth=0,
name=None,
description=None,
):
if not isinstance(max_subagent_depth, int) or max_subagent_depth < 0:
raise ValueError(
"`max_subagent_depth` must be a non-negative int, got "
f"{max_subagent_depth!r}"
)
# Domain attributes set before `super().__init__()`: the base
# constructor builds the (inherited) step + final generators from the
# instructions composed here, and `_get_builtin_tools` (called from
# there) needs none of them; RLM's callable tool is built per call.
self.max_subagent_depth = max_subagent_depth
self._subagent_depth = _subagent_depth
# Subagent delegation is offered only while we may still go one level
# deeper, so the deepest subagents can't fan out endlessly.
self._subagents_enabled = self._subagent_depth < self.max_subagent_depth
self.recursive = recursive
self.timeout = timeout
self.max_llm_calls = max_llm_calls
self.max_output_chars = max_output_chars
# `sub_language_model` defaults to the primary LM when omitted.
# ``get(None)`` would raise, so resolve only when a value is given.
self.sub_language_model = (
_get_lm(sub_language_model)
if sub_language_model is not None
else _get_lm(language_model)
)
# Sandbox handling: a concrete sandbox supplied at construction is
# reused across calls and its class becomes `sandbox_type`; otherwise a
# fresh `sandbox_type` (default MirageSandbox) is built per `call()`.
# Resolved here because the sandbox's prompt description is composed
# into the instructions below.
self.sandbox = sandbox
if sandbox is not None:
self.sandbox_type = type(sandbox)
else:
self.sandbox_type = sandbox_type or MirageSandbox
# Compose instructions before delegating to the base constructor. The
# final generator keeps the base instructions (recursive/default plus
# subagent guidance); only the step generator gets the sandbox
# description appended.
if not instructions:
if recursive:
instructions = get_recursive_instructions().replace(
"{max_llm_calls}", str(max_llm_calls)
)
else:
instructions = get_default_instructions()
if self._subagents_enabled:
# Idempotent: a serialized agent round-trips its post-append
# instructions back through __init__, so don't append twice.
guidance = get_subagent_tools_guidance()
if guidance not in instructions:
instructions = instructions + "\n\n" + guidance
resolved_final_instructions = final_instructions or instructions
sandbox_description = self.sandbox_type.description
if sandbox_description:
instructions = instructions + "\n\n" + sandbox_description
super().__init__(
schema=schema,
data_model=data_model,
language_model=language_model,
prompt_template=prompt_template,
examples=examples,
instructions=instructions,
final_instructions=resolved_final_instructions,
temperature=temperature,
max_tokens=max_tokens,
top_p=top_p,
top_k=top_k,
use_inputs_schema=use_inputs_schema,
use_outputs_schema=use_outputs_schema,
reasoning_effort=reasoning_effort,
use_chain_of_thought=use_chain_of_thought,
tools=[*(tools or []), *(sandbox_tools or [])] or None,
autonomous=autonomous,
return_inputs_with_trajectory=return_inputs_with_trajectory,
max_iterations=max_iterations,
streaming=streaming,
workdir=workdir,
skills=skills,
name=name,
description=description,
)
# `tools` / `sandbox_tools` are stored by the base constructor in
# `self.tools` but, for RLM, are exposed *inside* the sandbox
# (advertised via the catalog) rather than as native tool calls.
# Reject reserved helper names here, after the base's public-name
# check.
reserved = self._reserved_tool_names()
for tool_name in self.tools:
if tool_name in reserved:
raise ValueError(
f"Tool name '{tool_name}' is reserved by {type(self).__name__}."
)
# `native_tools` are the other half of the split: tools the LM calls
# directly, alongside `run_python_code`, instead of from inside a
# snippet. Sandbox placement is right for anything a snippet composes
# with (its result feeds the next line of Python), and wrong for a
# tool the LM just wants to *consult* — a lookup whose answer shapes
# the snippet it is about to write. Forcing the latter through the
# sandbox costs a round-trip through generated code, and the catalog
# advertises names the LM cannot call, which models routinely try
# anyway and get `Unknown tool` back.
self.native_tools = {}
for tool in native_tools or []:
if tool.name.startswith("_"):
raise ValueError(
f"Tool name {tool.name!r} starts with an underscore. "
f"Tools exposed to the LM must have public names; "
f"rename the function or pass an explicit `name=` to "
f"Tool(...)."
)
if tool.name in reserved:
raise ValueError(
f"Tool name '{tool.name}' is reserved by {type(self).__name__}."
)
if tool.name in self.tools:
raise ValueError(
f"Tool name '{tool.name}' is given as both a native tool "
"and a sandbox tool; a name can only be one or the other."
)
if tool.name in self.native_tools:
raise ValueError(f"Duplicate native tool name '{tool.name}'.")
self.native_tools[tool.name] = tool
# Only sandbox tools go in the catalog: native ones travel in the
# provider's own tool schema, so advertising them here would describe
# the same tool twice, in two different calling conventions.
self.tools_catalog = _build_tools_catalog(self.tools)
def _get_builtin_tools(self):
# RLM exposes no native tools at construction: its only callable tool,
# `run_python_code`, closes over a per-call sandbox and is built in
# `_begin_call`. User tools live inside the sandbox, not as native calls.
return []
def _requires_tools(self):
# `run_python_code` (built per call) is always available, so a user may
# construct an RLM agent with no tools at all.
return False
def _reserved_tool_names(self) -> frozenset:
"""Names a user tool cannot collide with at construction time.
``submit``, ``llm_query``, and ``llm_query_batched`` are always
reserved, even when ``recursive=False``, so that tool naming
stays stable across the two modes and a user tool can't quietly
shadow a helper name that would reappear if ``recursive`` is
flipped back on. The subagent helpers are reserved too, for the
same stability across ``max_subagent_depth`` settings.
"""
return frozenset(
{
"submit",
"llm_query",
"llm_query_batched",
"run_python_code",
"spawn_subagents",
"merge_subagent",
"discard_subagent",
}
)
def _build_extra_call_tools(self) -> dict:
"""Build the per-call recursive helpers when ``recursive=True``.
A fresh counter+lock is built on every invocation so concurrent
calls into the same agent instance get independent budgets.
Returns ``{}`` when ``recursive=False``.
"""
if not self.recursive:
return {}
counter = {"value": 0}
lock = asyncio.Lock()
return {
"llm_query": _build_llm_query_tool(
self.sub_language_model,
self.max_llm_calls,
counter,
lock,
),
"llm_query_batched": _build_llm_query_batched_tool(
self.sub_language_model,
self.max_llm_calls,
counter,
lock,
),
}
def _build_run_python_code_tool(self, sandbox):
"""Build the lone tool the LM can call.
``run_python_code`` is the *only* tool exposed to the LM. It wraps
the sandbox's ``run_python_code`` (which runs the
snippet in the persistent sandbox) and clips the captured streams
to ``max_output_chars``. The tools (``submit``, ``llm_query`` and
the user tools) and the ``inputs`` payload are not passed here;
`call` binds them onto the sandbox before each run (via
``bind_functions`` and a persisted ``inputs`` variable). The
sandbox is closed over, so a fresh tool is built per call. For
``compute_output_spec`` the closure is built with ``sandbox=None``
and never executed; only its signature/docstring drive the schema.
"""
max_output_chars = self.max_output_chars
def _clip(text):
if not text:
return ""
text = text.rstrip()
if max_output_chars is not None and len(text) > max_output_chars:
omitted = len(text) - max_output_chars
return text[:max_output_chars] + (
f"\n… (truncated, {omitted} chars omitted)"
)
return text
async def run_python_code(code: str) -> dict:
"""Execute one Python snippet in the persistent sandbox.
State persists across calls: variables, imports and function
definitions stay defined. The user input is bound as a dict named
`inputs`; read full values via `inputs[field]`. Other tools
(`submit`, `llm_query`, ...) are pre-imported functions; call them
directly, e.g. `out = llm_query(prompt)`. Call `submit(result={...})`
to end the run.
Args:
code (str): The Python snippet to execute in the
sandbox.
"""
result = await sandbox.run_python_code(code)
observation = {"stdout": _clip(result.get("stdout", ""))}
stderr = _clip(result.get("stderr", ""))
if stderr:
observation["stderr"] = stderr
if result.get("error"):
observation["error"] = result["error"]
return observation
return Tool(run_python_code, name="run_python_code")
def _build_subagent_tools(self, sandbox, registry, counter, repl_state):
"""Build the per-call subagent tools (spawn / merge / discard).
These are *native* tools the LM calls directly, never from inside a
``run_python_code`` snippet, so the sandbox REPL is idle when they
fork or merge it (a busy REPL can't be dumped). The closures capture
the resolved ``sandbox``, a per-call ``registry`` (handle -> fork),
a handle ``counter``, and ``repl_state`` tracking the single REPL
adoption allowed per turn.
"""
from synalinks.src.modules.core.input_module import Input
from synalinks.src.programs.program import Program
async def spawn_subagents(tasks: List[str]) -> dict:
"""Run subagents in parallel, each on an isolated fork of the sandbox.
Each task is handed to a fresh subagent on its own fork that
inherits your current REPL state (variables, functions, imports)
and files. Subagents run concurrently; their changes are isolated.
Review each returned ``patch`` then ``merge_subagent(handle)`` to
fold a subagent's work into your sandbox.
Args:
tasks (list): One instruction string per subagent describing
what that subagent should accomplish.
Returns:
dict: ``subagents``: a list of ``{handle, task, result,
diff, patch}`` per subagent, where ``patch`` is the subagent's
pending changes as a git-style unified diff (the actual
line-level edits) and ``diff`` is the structured
``{written, deleted}`` summary; or ``{handle, task, error}``
for a failed subagent; or a top-level ``error`` when ``tasks``
is empty.
"""
prompts = [str(t) for t in (tasks or [])]
if not prompts:
return {"error": "no tasks provided"}
async def run_one(index, prompt):
# Subagents inherit the parent's confinement (``confine=None``):
# when this sandbox is confined, the subagent is confined to its
# OWN fork (host hidden, network cut, isolated filesystem, and
# the parent's egress/mount/seccomp posture); when the parent
# runs unconfined, so does the subagent.
fork = sandbox.fork(
copy_repl=True, name=f"{self.name}_sub{index}", confine=None
)
subagent = RecursiveLanguageModelAgent(
language_model=self.sub_language_model,
sub_language_model=self.sub_language_model,
sandbox=fork,
recursive=self.recursive,
instructions=get_subagent_instructions(),
temperature=self.temperature,
max_tokens=self.max_tokens,
top_p=self.top_p,
top_k=self.top_k,
reasoning_effort=self.reasoning_effort,
use_chain_of_thought=self.use_chain_of_thought,
max_iterations=self.max_iterations,
max_llm_calls=self.max_llm_calls,
max_output_chars=self.max_output_chars,
max_subagent_depth=self.max_subagent_depth,
_subagent_depth=self._subagent_depth + 1,
return_inputs_with_trajectory=False,
autonomous=True,
name=f"{self.name}_sub{index}",
)
# Run through a Program (the canonical path) so the subagent's
# build is LM-free and it isn't double-invoked.
inp = Input(data_model=ChatMessages)
out = await subagent(inp)
program = Program(
inputs=inp, outputs=out, name=f"{self.name}_sub{index}_prog"
)
messages = ChatMessages(
messages=[ChatMessage(role=ChatRole.USER, content=prompt)]
)
result = await program(messages)
return fork, _subagent_answer(result)
results = await asyncio.gather(
*(run_one(i, p) for i, p in enumerate(prompts)),
return_exceptions=True,
)
report = []
for prompt, res in zip(prompts, results):
handle = f"subagent_{counter[0]}"
counter[0] += 1
if isinstance(res, Exception):
report.append(
{
"handle": handle,
"task": prompt,
"error": f"{type(res).__name__}: {res}",
}
)
continue
fork, answer = res
registry[handle] = fork
report.append(
{
"handle": handle,
"task": prompt,
"result": answer,
"diff": fork.diff(),
"patch": fork.patch(),
}
)
return {"subagents": report}
async def merge_subagent(
handle: str,
paths: Optional[List[str]] = None,
force: bool = False,
adopt_repl: bool = False,
) -> dict:
"""Fold a subagent's changes into your sandbox.
Merges the subagent's file changes (``paths`` / ``force`` as for
files). With ``adopt_repl=True``, also adopts the subagent's whole
Python namespace (variables/functions/imports); only ONE
subagent's REPL can be adopted per ``spawn_subagents`` batch (a
second would overwrite the first).
Args:
handle (str): A handle returned by ``spawn_subagents``.
paths (list): Optional subset of virtual paths to merge.
force (bool): Apply conflicting file paths instead of
refusing them.
adopt_repl (bool): Also adopt the subagent's whole REPL
namespace.
Returns:
dict: ``{written, deleted, conflicts, skipped, repl_adopted}``,
or ``error`` for an unknown handle.
"""
fork = registry.get(handle)
if fork is None:
return {"error": f"unknown subagent handle: {handle!r}"}
do_repl = bool(adopt_repl)
note = None
if do_repl and repl_state["adopted"]:
do_repl = False
note = (
"REPL already adopted from another subagent this turn; "
"merged files only (a second adoption would overwrite the "
"first)."
)
report = sandbox.merge(fork, paths=paths, force=force, repl=do_repl)
if do_repl:
repl_state["adopted"] = True
if note:
report = dict(report)
report["repl_warning"] = note
return report
async def discard_subagent(handle: str) -> dict:
"""Drop a subagent's fork without applying any of its changes.
Args:
handle (str): A handle returned by ``spawn_subagents``.
Returns:
dict: ``{discarded: handle}``, or ``error`` for an unknown
handle.
"""
if registry.pop(handle, None) is None:
return {"error": f"unknown subagent handle: {handle!r}"}
return {"discarded": handle}
return {
"spawn_subagents": Tool(spawn_subagents, name="spawn_subagents"),
"merge_subagent": Tool(merge_subagent, name="merge_subagent"),
"discard_subagent": Tool(discard_subagent, name="discard_subagent"),
}
async def _begin_call(self, inputs, training, *, sandbox=None):
# Per-call tool set: user tools plus a fresh `submit` bound to a
# private holder, plus any per-call recursive helpers. submit is the
# canonical termination signal, always exposed, schema'd or not, and
# everything here is built fresh per call so concurrent invocations
# don't share holders, counters, or locks.
call_tools = dict(self.tools)
submit_holder = {"value": None}
call_tools["submit"] = _build_submit_tool(self.schema, submit_holder)
call_tools.update(self._build_extra_call_tools())
call_tools_catalog = _build_tools_catalog(call_tools)
if is_chat_messages(inputs):
trajectory = inputs
inputs_json = {}
else:
inputs_json = inputs.get_json()
# The LM prompt only sees a metadata summary of the inputs:
# previews and sizes, never the full value. The sandbox gets the
# complete `inputs_json` rebound on every `run_python_code` call,
# so `inputs[field]` is always reachable.
base = summarize_inputs(inputs_json)
if call_tools_catalog is not None:
base = await ops.concat(
base,
call_tools_catalog,
name="inputs_with_tools_" + self.name,
)
trajectory = await ops.concat(
base,
ChatMessages(),
name="trajectory_" + self.name,
)
# Sandbox resolution order: per-call kwarg > constructor-supplied
# sandbox > fresh sandbox of `sandbox_type`. The first two cases let the
# caller (or the agent's owner) keep REPL state alive across calls; the
# third is the stateless-per-call default. A freshly built sandbox is
# seeded from `workdir` when one is set.
if sandbox is None:
if self.sandbox is not None:
sandbox = self.sandbox
elif self.workdir is not None:
sandbox = self.sandbox_type(workdir=self.workdir, timeout=self.timeout)
else:
sandbox = self.sandbox_type(timeout=self.timeout)
# The per-turn snippet is delivered as a native `run_python_code` tool
# call, the only tool the LM can call, wrapping the sandbox's own
# `run_python_code`. The sandbox-side tools (submit, llm_query, user
# tools) are NOT exposed to the LM; they live inside the sandbox as
# plain synchronous functions (advertised via the tools catalog). Bind
# them onto the sandbox so every `run_python_code` snippet can reach them.
external_functions = {
name: _adapt_tool_for_sandbox(t) for name, t in call_tools.items()
}
sandbox.bind_functions(external_functions)
# Persist the full input payload as `inputs` in the sandbox namespace.
# The per-run `inputs=` binding does not persist, so copy it into a real
# variable once; every snippet then reads it via `inputs[field]`.
await sandbox.run("inputs = _rlm_inputs", inputs={"_rlm_inputs": inputs_json})
run_tool = self._build_run_python_code_tool(sandbox)
# Subagent tools (spawn / merge / discard) are *native* tools the LM
# calls directly alongside run_python_code, never from inside a
# snippet, so the REPL is idle when they fork/merge it. Built fresh per
# call with a private fork registry and a single-REPL-adoption guard.
subagent_registry = {}
extra_native_tools = {}
if self._subagents_enabled:
extra_native_tools = self._build_subagent_tools(
sandbox, subagent_registry, [0], {"adopted": False}
)
# User-supplied `native_tools` join them: same dispatch path, same
# provider schema, same `Unknown tool` message listing what is
# callable. They are deliberately absent from `call_tools`, so they
# are neither bound into the sandbox namespace nor advertised in the
# catalog as snippet-reachable.
extra_native_tools = {**extra_native_tools, **self.native_tools}
ctx = {
"sandbox": sandbox,
"run_tool": run_tool,
"extra_native_tools": extra_native_tools,
"submit_holder": submit_holder,
"native_tools": [run_tool, *extra_native_tools.values()],
"submitted_final": None,
}
return trajectory, ctx
def _native_tools(self, ctx):
# The LM calls `run_python_code`, the subagent tools, and any tool
# passed as `native_tools=`; everything in `tools=` / `sandbox_tools=`
# is reachable from inside the sandbox instead, not as a native call.
return ctx["native_tools"]
def _requires_tools(self):
return False
def _on_empty_generation(self, agent_messages, ctx):
# RLM simply stops the loop on an empty generation (no message added).
return None
async def _on_no_tool_calls(self, tool_calls, agent_messages, ctx):
# The shared loop only appends the assistant message when there are
# native tool calls; on an empty turn RLM keeps it and nudges the model
# toward `run_python_code` / `submit`, then keeps iterating.
agent_messages.append(tool_calls.get_json())
agent_messages.append(
ChatMessage(
role=ChatRole.USER,
content=(
"(no tool call) Call `run_python_code` with a snippet, and "
"call `submit(result={...})` inside it to terminate the run."
),
).get_json()
)
return False
async def _dispatch_tool_calls(self, native_tool_calls, agent_messages, ctx):
# Sequential dispatch: a `submit()` inside a snippet writes to the
# holder, short-circuiting the run. The subagent tools fork/merge the
# idle REPL, so they must not run concurrently with a snippet.
run_tool = ctx["run_tool"]
extra_native_tools = ctx["extra_native_tools"]
submit_holder = ctx["submit_holder"]
for tool_call in native_tool_calls:
function = tool_call.get("function") or {}
tool_name = function.get("name")
tool_args = function.get("arguments") or {}
tool_call_id = tool_call.get("id")
if tool_name in extra_native_tools:
# spawn_subagents / merge_subagent / discard_subagent:
# native tools, invoked with the REPL idle.
try:
result = await extra_native_tools[tool_name](**tool_args)
content = result.get_json() if hasattr(result, "get_json") else result
except Exception as e:
content = {"error": f"{type(e).__name__}: {e}"}
elif tool_name != "run_python_code":
callable_tools = ", ".join(
["'run_python_code'", *(f"'{n}'" for n in extra_native_tools)]
)
content = {
"error": (
f"Unknown tool '{tool_name}'. Callable tools are "
f"{callable_tools}; everything else is a sandbox "
"function reachable from your snippet."
)
}
else:
try:
result = await run_tool(**tool_args)
content = result.get_json() if hasattr(result, "get_json") else result
except Exception as e:
content = {"error": f"{type(e).__name__}: {e}"}
# submit() inside the snippet wrote to the holder. Clear it
# either way so a stale payload can't short-circuit a later retry.
submitted = submit_holder["value"]
submit_holder["value"] = None
if submitted is not None:
content = dict(content)
if self.schema:
try:
jsonschema.validate(submitted, self.schema)
except ValidationError as ve:
content["submit"] = (
f"validation failed: {ve.message}. "
"Revise the payload and call submit again."
)
else:
ctx["submitted_final"] = submitted
content["submit"] = "accepted"
else:
ctx["submitted_final"] = submitted
content["submit"] = "accepted"
agent_messages.append(
ChatMessage(
role=ChatRole.TOOL,
tool_call_id=tool_call_id,
content=content,
).get_json()
)
if ctx["submitted_final"] is not None:
return True
return False
async def _run_interactive(self, trajectory, agent_messages, ctx, training):
# RLM's interactive mode is a single pass of the same loop.
return await self._run_loop(
trajectory, agent_messages, ctx, training, max_steps=1
)
def _wrap_trajectory(self, agent_messages):
return JsonDataModel(
json=ChatMessages(
messages=[ChatMessage(**msg) for msg in agent_messages]
).get_json(),
schema=ChatMessages.get_schema(),
name=self.name,
)
async def _finish(self, trajectory, agent_messages, ctx, training):
submitted_final = ctx["submitted_final"]
# Interactive mode: only invoke the final generator when the LM itself
# signalled completion via submit. Otherwise return the updated
# trajectory so the caller can decide when to continue.
if not self.autonomous and submitted_final is None:
return self._wrap_trajectory(agent_messages)
# submit short-circuit: the LM already produced the final payload inside
# the sandbox, so skip the final-formatting LM call. Schemaless mode
# treats the payload as the content of a final assistant message.
if submitted_final is not None:
if self.schema:
final_result = JsonDataModel(
json=submitted_final,
schema=self.schema,
name="final_generator_" + self.name,
)
else:
agent_messages.append(
ChatMessage(
role=ChatRole.ASSISTANT,
content=submitted_final.get("answer", submitted_final),
).get_json()
)
return self._wrap_trajectory(agent_messages)
else:
final_result = await self.final_generator(trajectory)
if not self.schema:
# Schemaless fallback: the final generator emits a ChatMessage.
if final_result:
agent_messages.append(final_result.get_json())
return self._wrap_trajectory(agent_messages)
if self.return_inputs_with_trajectory:
return await ops.concat(
self._wrap_trajectory(agent_messages),
final_result,
name=self.name,
)
return final_result
async def compute_output_spec(self, inputs, training=False, sandbox=None, **kwargs):
# See FunctionCallingAgent.compute_output_spec: `call()` takes **kwargs
# (sandbox is threaded through it), so mirror the signature here.
if not self.autonomous and not is_chat_messages(inputs):
raise ValueError(
f"In interactive mode, the {type(self).__name__} needs a "
"ChatMessages-like data model as inputs"
)
# Mirror the runtime: the code generator sees a summary of the
# input plus the tool catalog, not the raw input DataModel.
if is_chat_messages(inputs):
generator_inputs = inputs
else:
generator_inputs = SymbolicDataModel(
schema=InputsSummary.get_schema(),
name="inputs_summary_" + self.name,
)
if self.tools_catalog is not None:
generator_inputs = await ops.concat(
generator_inputs,
self.tools_catalog,
name="inputs_with_tools_" + self.name,
)
# The closure placeholders are never executed during spec tracing;
# only the tool's signature/docstring shape the prompt.
spec_tool = self._build_run_python_code_tool(None)
_ = await self.tool_calls_generator(generator_inputs, tools=[spec_tool])
if not self.autonomous:
# Interactive mode: the common case is returning the trajectory.
# When the LM emits empty code the runtime returns the final
# answer instead; we pick the common-case spec.
return SymbolicDataModel(
schema=ChatMessages.get_schema(),
name=self.name,
)
if not self.schema:
# Schemaless autonomous: final generator produces a ChatMessage
# appended to the trajectory; output spec is ChatMessages.
_ = await self.final_generator(inputs)
return SymbolicDataModel(
schema=ChatMessages.get_schema(),
name=self.name,
)
if self.return_inputs_with_trajectory:
return await ops.logical_and(
SymbolicDataModel(
schema=ChatMessages.get_schema(),
name=self.name,
),
SymbolicDataModel(
schema=self.schema,
name="final_generator_" + self.name,
),
name=self.name,
)
return await self.final_generator(inputs)
def get_config(self):
config = super().get_config()
config.update(
{
"recursive": self.recursive,
"timeout": self.timeout,
"max_llm_calls": self.max_llm_calls,
"max_output_chars": self.max_output_chars,
"max_subagent_depth": self.max_subagent_depth,
"sandbox_type": get_registered_name(self.sandbox_type),
"sub_language_model": serialization_lib.serialize_synalinks_object(
self.sub_language_model,
),
"sandbox": (
serialization_lib.serialize_synalinks_object(self.sandbox)
if self.sandbox is not None
else None
),
}
)
return config
@classmethod
def from_config(cls, config):
config = dict(config)
if config.get("sub_language_model") is not None:
config["sub_language_model"] = serialization_lib.deserialize_synalinks_object(
config.pop("sub_language_model")
)
else:
config.pop("sub_language_model", None)
sandbox_type_name = config.pop("sandbox_type", None)
config["sandbox_type"] = (
get_registered_object(sandbox_type_name) if sandbox_type_name else None
)
if config.get("sandbox") is not None:
config["sandbox"] = serialization_lib.deserialize_synalinks_object(
config.pop("sandbox")
)
else:
config.pop("sandbox", None)
return super().from_config(config)